package com.orkasgb.framework.hadoop.mapperreduce.join.reducejoin;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class JoinMapper extends Mapper<LongWritable, Text, Text, JoinEntity> {

    private final Text text = new Text();

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, JoinEntity>.Context context) throws IOException, InterruptedException {
        if (key.get() == 0) {
            return;
        }

        String[] split = value.toString().split("\t");
        JoinEntity joinEntity = new JoinEntity();
        FileSplit fileSplit = (FileSplit) context.getInputSplit();
        String filename = fileSplit.getPath().getName();
        String id = split[1];
        joinEntity.setNo(Integer.parseInt(split[0]));
        joinEntity.setId(id);
        if (filename.equals("order.txt")) {
            joinEntity.setNum(Integer.parseInt(split[2]));
            joinEntity.setDate(split[3]);

            joinEntity.setName("");
            joinEntity.setflag("order");
        } else {
            joinEntity.setNum(0);
            joinEntity.setDate("");

            joinEntity.setName(split[2]);
            joinEntity.setflag("product");
        }

        text.set(id);
        context.write(text, joinEntity);

    }
}
